package cn.doitedu.api;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

public class _33_Exercise_OurLeftJoin {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        //env.getCheckpointConfig().setTolerableCheckpointFailureNumber();

        // {"order_id":"o1","order_amount":3000}
        DataStream<OrderMain> stream1 = env.
                socketTextStream("doitedu", 8891)
                .map(s -> JSON.parseObject(s, OrderMain.class));

        // o1,p1,1200
        // o1,p2,1800
        DataStream<OrderItem> stream2 = env
                .socketTextStream("doitedu", 8892)
                .map(s -> JSON.parseObject(s, OrderItem.class));


        // 实现left join的效果
        // 利用  connect  + keyBy  + process
        stream1.connect(stream2)
                .keyBy(OrderMain::getOrderId, OrderItem::getOrderId)
                .process(new KeyedCoProcessFunction<String, OrderMain, OrderItem, String>() {

                    ListState<OrderMain> left;
                    ListState<OrderItem> right;

                    ValueState<Integer> flagState;


                    @Override
                    public void open(Configuration parameters) throws Exception {

                        left = getRuntimeContext().getListState(new ListStateDescriptor<OrderMain>("left", OrderMain.class));
                        right = getRuntimeContext().getListState(new ListStateDescriptor<OrderItem>("right", OrderItem.class));

                        // 这个状态是用来记录是否左表数据输出过带null结果
                        flagState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("flag", Integer.class));

                    }

                    @Override
                    public void processElement1(OrderMain orderMain, KeyedCoProcessFunction<String, OrderMain, OrderItem, String>.Context ctx, Collector<String> out) throws Exception {

                        JSONObject resultObject = new JSONObject();

                        boolean joined = false;

                        for (OrderItem orderItem : right.get()) {
                            resultObject.put("before", null);

                            JSONObject dataObject = new JSONObject();
                            dataObject.put("order_id", orderMain.getOrderId());
                            dataObject.put("order_amount", orderMain.getOrderAmount());
                            dataObject.put("product_id", orderItem.getProductId());
                            dataObject.put("product_amount", orderItem.getProductAmount());
                            resultObject.put("after", dataObject);


                            resultObject.put("row_kind", "+I");

                            // 输出关联结果
                            out.collect(resultObject.toJSONString());

                            joined = true;

                            // 标记 输出过null的 标记为0 （表示已经不需要做特别处理了）
                            flagState.update(0);

                        }

                        if (!joined) {
                            resultObject.put("before", null);

                            JSONObject dataObject = new JSONObject();

                            dataObject.put("order_id", orderMain.getOrderId());
                            dataObject.put("order_amount", orderMain.getOrderAmount());
                            dataObject.put("product_id", null);
                            dataObject.put("product_amount", null);
                            resultObject.put("after", dataObject);

                            resultObject.put("row_kind", "+I");

                            // 标记 输出null的 标记为1 （表示右表要做特别处理）
                            flagState.update(1);

                            // 输出关联结果(带null的结果)
                            out.collect(resultObject.toJSONString());

                        }


                        left.add(orderMain);


                    }

                    @Override
                    public void processElement2(OrderItem orderItem, KeyedCoProcessFunction<String, OrderMain, OrderItem, String>.Context ctx, Collector<String> out) throws Exception {

                        JSONObject resultObject = new JSONObject();

                        right.add(orderItem);

                        OrderMain orderMain = null;
                        for (OrderMain tmpMain : left.get()) {
                            orderMain = tmpMain;
                        }

                        if (orderMain == null) return;

                        JSONObject after = new JSONObject();
                        after.put("order_id", orderMain.getOrderId());
                        after.put("order_amount", orderMain.getOrderAmount());
                        after.put("product_id", orderItem.getProductId());
                        after.put("product_amount", orderItem.getProductAmount());

                        resultObject.put("after", after);
                        resultObject.put("row_kind", "+I");


                        // 判断标记，来决定是否需要 带上  before
                        JSONObject before = new JSONObject();
                        if (flagState.value() != null && flagState.value() == 1) {
                            before.put("order_id", orderMain.getOrderId());
                            before.put("order_amount", orderMain.getOrderAmount());
                            before.put("product_id", null);
                            before.put("product_amount", null);

                            resultObject.put("before", before);

                            resultObject.put("row_kind", "U");
                        }

                        // 取消 特别 标记
                        flagState.update(0);

                        int x = flagState.value();


                        // 输出本次关联结果
                        out.collect(resultObject.toJSONString());

                    }


                }).print();

        env.execute();


    }
}
